--- title: Java 并发之分工工具 date: 2020-07-14 15:27:46 categories: - Java - JavaCore - 并发 tags: - Java - JavaCore - 并发 - FutureTask - CompletableFuture - CompletionStage - CompletionService - ForkJoinPool permalink: /pages/8b685389/ --- # Java 并发之分工工具 ## 简介 Java 并发编程中，**分工工具**是处理复杂并行任务的核心利器。当单个任务过于庞大或具有复杂的依赖关系时，我们需要将其拆分为多个子任务并行执行，最后汇总结果。Java 提供了多种分工工具来应对不同场景：`ForkJoinPool` 面向分治递归任务，`CompletableFuture` 面向异步编排与聚合，`CompletionService` 面向批量并行任务的结果收集。 **对于简单的并行任务，你可以通过"线程池 + Future"的方案来解决；如果任务之间有聚合关系，无论是 AND 聚合还是 OR 聚合，都可以通过 CompletableFuture 来解决；而批量的并行任务，则可以通过 CompletionService 来解决。** ## ForkJoinPool ForkJoinPool 是 Java 7 引入的一种线程池，专为**分治任务（Divide-and-Conquer）** 设计，核心思想是将大任务拆分为多个小任务（Fork），并行执行后合并结果（Join）。它在处理可分解的复杂任务时效率显著，尤其适合计算密集型场景。 **ForkJoinPool 通过工作窃取机制高效处理分治任务，适合递归并行计算，核心是本地队列+LIFO 处理+FIFO 窃取。** ![](https://raw.githubusercontent.com/dunwu/images/master/archive/2020/07/338c79b1b1d34e258454af64336b9356.png) ### ForkJoinPool 特性 **关键特性** - **工作窃取算法**：空闲线程从繁忙线程的任务队列中 "窃取" 任务执行，减少线程竞争，提高 CPU 利用率。 - **分治递归**：适合处理可递归拆分的计算密集型任务 - **并行处理**：默认并行度为 CPU 核心数，可自定义；提供公共池（`commonPool`）供全局使用，减少资源消耗。 - **任务拆分**：大任务自动拆分为小任务，直到达到阈值 **与普通线程池对比** | **特性** | **ForkJoinPool** | **ThreadPoolExecutor** | | ------------ | ------------------------- | ---------------------- | | **任务类型** | 分治任务（递归拆分） | 独立任务 | | **任务调度** | 任务窃取（本地队列+窃取） | 全局队列（可能竞争） | | **适用场景** | CPU 密集型并行计算 | IO 密集型或短任务 | ### ForkJoinPool 原理 - **双端队列**：每个工作线程（`ForkJoinWorkerThread`）维护自己的任务队列 - **任务调度**： - **拆分（fork）**：调用 `fork()` 将子任务加入当前线程队列。 - **合并（join）**：调用 `join()` 等待任务完成，必要时帮助执行任务。 - **窃取（stealing）**：若线程无任务，从其他线程队列**尾部窃取任务**（FIFO，减少竞争）。 ### ForkJoinPool 用法 - **定义任务**：`ForkJoinTask` 是所有 `ForkJoin` 任务的父类。 - 继承 `RecursiveTask`（有返回值）或 `RecursiveAction`（无返回值）。 - 重写 `compute()` 方法：若任务小于阈值则直接计算，否则分解为子任务。 - **任务调度**：`ForkJoinPool` 是线程池的核心实现类。 - 用 `fork()` 异步提交子任务到线程池。 - 用 `join()` 阻塞等待子任务结果并合并。 - 通过 `invoke()`（同步执行）或`submit()`（异步执行）提交根任务。 ### ForkJoinPool 示例 ::: tabs @tab 计算斐波那契数列 ```java public static void main(String[] args) { long number = 10; ForkJoinPool pool = new ForkJoinPool(); FibonacciTask task = new FibonacciTask(number); long result = pool.invoke(task); System.out.println("斐波那契数列第 " + number + " 项：" + result); } public static class FibonacciTask extends RecursiveTask { private final long N; public FibonacciTask(long n) { this.N = n; } @Override protected Long compute() { if (N <= 1) { return N; } // Fork：分解任务 FibonacciTask f1 = new FibonacciTask(N - 1); FibonacciTask f2 = new FibonacciTask(N - 2); // 异步执行第一个任务 f1.fork(); // 同步执行第二个任务并获取结果 long result2 = f2.compute(); // Join：合并结果 long result1 = f1.join(); return result1 + result2; } } ``` @tab 数组求和 ```java public static void main(String[] args) { // 创建一个随机数组 int[] array = new int[10000]; for (int i = 0; i < array.length; i++) { array[i] = i + 1; } // 使用 ForkJoinPool 计算 ForkJoinPool pool = new ForkJoinPool(); ArraySum task = new ArraySum(array); int result = pool.invoke(task); System.out.println("数组总和：" + result); // 验证结果 int expected = Arrays.stream(array).sum(); System.out.println("验证结果：" + expected); System.out.println("结果是否正确：" + (result == expected)); } public static class ArraySum extends RecursiveTask { private static final int THRESHOLD = 1000; // 阈值，小于此值时直接计算 private final int[] array; private final int start; private final int end; public ArraySum(int[] array) { this(array, 0, array.length); } private ArraySum(int[] array, int start, int end) { this.array = array; this.start = start; this.end = end; } @Override protected Integer compute() { int length = end - start; // 如果任务足够小，直接计算 if (length <= THRESHOLD) { int sum = 0; for (int i = start; i < end; i++) { sum += array[i]; } return sum; } // 否则分解任务 int mid = start + (end - start) / 2; ArraySum leftTask = new ArraySum(array, start, mid); ArraySum rightTask = new ArraySum(array, mid, end); // 执行子任务 leftTask.fork(); int rightResult = rightTask.compute(); int leftResult = leftTask.join(); // 合并结果 return leftResult + rightResult; } } ``` @tab 模拟 MapReduce 统计单词数量 ```java static void main(String[] args) { String[] fc = { "hello world", "hello me", "hello fork", "hello join", "fork join in world" }; //创建 ForkJoin 线程池 ForkJoinPool fjp = new ForkJoinPool(3); //创建任务 MR mr = new MR(fc, 0, fc.length); //启动任务 Map result = fjp.invoke(mr); //输出结果 result.forEach((k, v) -> System.out.println(k + ":" + v)); } //MR 模拟类 static class MR extends RecursiveTask> { private String[] fc; private int start, end; //构造函数 MR(String[] fc, int fr, int to) { this.fc = fc; this.start = fr; this.end = to; } @Override protected Map compute() { if (end - start == 1) { return calc(fc[start]); } else { int mid = (start + end) / 2; MR mr1 = new MR(fc, start, mid); mr1.fork(); MR mr2 = new MR(fc, mid, end); //计算子任务，并返回合并的结果 return merge(mr2.compute(), mr1.join()); } } //合并结果 private Map merge(Map r1, Map r2) { Map result = new HashMap<>(); result.putAll(r1); //合并结果 r2.forEach((k, v) -> { Long c = result.get(k); if (c != null) { result.put(k, c + v); } else { result.put(k, v); } }); return result; } //统计单词数量 private Map calc(String line) { Map result = new HashMap<>(); //分割单词 String[] words = line.split("\\s+"); //统计单词数量 for (String w : words) { Long v = result.get(w); if (v != null) { result.put(w, v + 1); } else { result.put(w, 1L); } } return result; } } ``` ::: ## FutureTask FutureTask 有两个构造函数： ```java FutureTask(Callable callable); FutureTask(Runnable runnable, V result); ``` `FutureTask` 实现了 `Runnable` 和 `Future` 接口。由于实现了 `Runnable` 接口，所以可以将 `FutureTask` 对象作为任务提交给 `ThreadPoolExecutor` 去执行，也可以直接被 `Thread` 执行；又因为实现了 `Future` 接口，所以也能用来获得任务的执行结果。下面，通过一组示例来展示 FutureTask 如何分别交给线程池、线程执行。 ::: tabs#创建 FutureTask 示例 @tab `FutureTask` 交给线程池执行【示例】`FutureTask` 交给线程池执行 ```java public class FutureTaskDemo { public static void main(String[] args) throws ExecutionException, InterruptedException { // 创建 FutureTask Task task = new Task(); FutureTask f1 = new FutureTask<>(task); FutureTask f2 = new FutureTask<>(task); // 创建线程池 ExecutorService executor = Executors.newCachedThreadPool(); executor.submit(f1); executor.submit(f2); System.out.println(f1.get()); System.out.println(f2.get()); executor.shutdown(); } static class Task implements Callable { @Override public String call() { return Thread.currentThread().getName() + " 执行成功！"; } } } // 输出 // pool-1-thread-1 执行成功！ // pool-1-thread-2 执行成功！ ``` @tab `FutureTask` 交给线程执行【示例】`FutureTask` 交给线程执行 ```java public class FutureTaskDemo2 { public static void main(String[] args) throws InterruptedException, ExecutionException { // 创建 FutureTask Task task = new Task(); FutureTask f1 = new FutureTask<>(task); FutureTask f2 = new FutureTask<>(task); // 创建线程 new Thread(f1).start(); new Thread(f2).start(); System.out.println(f1.get()); System.out.println(f2.get()); } static class Task implements Callable { @Override public String call() { return Thread.currentThread().getName() + " 执行成功！"; } } } // 输出 // Thread-0 执行成功！ // Thread-1 执行成功！ ``` @tab 用 `FutureTask` 完成并行计算【示例】用 `FutureTask` 完成并行计算 ```java public class FutureTaskDemo3 { public static void main(String[] args) throws InterruptedException, ExecutionException { // 创建一个线程池来执行任务 ExecutorService executor = Executors.newFixedThreadPool(2); // 创建两个 Callable 对象 Callable t1 = () -> { int result = 0; for (int i = 1; i <= 100; i++) { result += i; } return result; }; Callable t2 = () -> { int result = 0; for (int i = 101; i <= 200; i++) { result += i; } return result; }; // 创建两个 FutureTask 对象 FutureTask f1 = new FutureTask<>(t1); FutureTask f2 = new FutureTask<>(t2); // 提交任务到线程池执行 executor.execute(f1); executor.execute(f2); // 获取任务的结果 Integer value1 = f1.get(); Integer value2 = f2.get(); System.out.println("total = " + value1 + value2); // 关闭线程池 executor.shutdown(); } } ``` ::: ## CompletableFuture JDK8 提供了 CompletableFuture 来支持异步编程。 ### CompletableFuture 特性 - **链式调用**：通过`thenApply()`、`thenAccept()`、`thenCompose()`等方法实现任务流水线 - **组合操作**：提供`allOf()`（等待所有任务完成）、`anyOf()`（等待任一任务完成）等方法，支持复杂任务依赖管理 - **异常处理**：通过`exceptionally()`、`handle()`等方法统一处理异步任务中的异常，无需 try-catch 嵌套 - **线程池灵活配置**：默认使用`ForkJoinPool.commonPool()`，也可指定自定义线程池，控制任务执行线程 ### CompletableFuture 原理 CompletableFuture 通过**状态机**管理任务生命周期，**回调列表**实现链式依赖，**线程池**调度异步执行，最终实现高效的非阻塞异步编程。其核心是将任务、依赖关系、线程调度解耦，通过 CAS 和 `volatile` 保证线程安全，同时避免回调嵌套问题。 #### 状态管理 - 内部维护任务状态（`enum` 类型）：`NEW`（初始）、`COMPLETING`（完成中）、`NORMAL`（正常完成）、`EXCEPTIONAL`（异常完成）、`CANCELLED`（取消）等。 - 通过 `volatile` 变量 `result` 存储计算结果或异常，确保线程间可见性。 - 状态转换通过 CAS 操作保证原子性，例如任务完成时从 `NEW` 转为 `NORMAL` 并设置结果。 #### 回调链触发机制 - **回调注册**：`thenApply`/`thenAccept` 等方法会创建新的 `CompletableFuture`（依赖原任务），并将回调函数（如 `Function`/`Consumer`）注册到原任务的回调列表中。 - **触发时机**：当原任务完成（状态变为 `NORMAL` 或 `EXCEPTIONAL`），会遍历回调列表，在**当前线程**或**指定线程池**中执行回调： - 若原任务由线程池线程完成，回调默认由该线程继续执行（避免线程切换开销）。 - 若通过 `thenApplyAsync` 等带 `Async` 的方法，回调会提交到新的线程池执行。 - **链式传递**：回调执行结果会作为新 `CompletableFuture` 的结果，继续触发后续回调，形成流水线。 #### 任务执行与线程调度 - **异步任务提交**：`supplyAsync`/`runAsync` 会将任务包装为 `AsyncSupply`/`AsyncRun`（实现 `Runnable`），提交到指定线程池（默认 `ForkJoinPool.commonPool()`）。 - **线程池执行**：线程池工作线程执行任务，完成后调用 `complete()` 或 `completeExceptionally()` 更新状态并触发回调。 #### 多任务协同（`allOf`/`anyOf`） - **`allOf`**：内部维护计数器，记录未完成的依赖任务数。每个依赖任务完成时计数器减 1，当计数器为 0 时，`allOf` 返回的 `CompletableFuture` 完成。 - **`anyOf`**：监听所有依赖任务，当第一个任务完成（正常或异常），`anyOf` 返回的 `CompletableFuture` 立即以该结果完成。 ### CompletableFuture 用法 #### 创建 CompletableFuture 常见的创建 `CompletableFuture` 对象的方法如下： - 通过 `new` 关键字 - 静态工厂方法：`runAsync()`、`supplyAsync()` - `runAsync(Runnable)`：无返回值的异步任务 - `supplyAsync(Supplier)`：有返回值的异步任务 ::: tabs @tab new CompletableFuture 示例 ```java CompletableFuture> resultFuture = new CompletableFuture<>(); ``` @tab CompletableFuture 静态工厂方法示例 ```java // 无返回值的异步任务 CompletableFuture future1 = CompletableFuture.runAsync(() -> { System.out.println("异步执行"); }); // 有返回值的异步任务 CompletableFuture future2 = CompletableFuture.supplyAsync(() -> { return "计算结果"; }); // 指定自定义线程池 ExecutorService executor = Executors.newFixedThreadPool(4); CompletableFuture future3 = CompletableFuture.supplyAsync(() -> { return "使用自定义线程池"; }, executor); ``` ::: 默认情况下 `CompletableFuture` 会使用公共的 `ForkJoinPool` 线程池，这个线程池默认创建的线程数是 CPU 的核数（也可以通过 JVM option: `-Djava.util.concurrent.ForkJoinPool.common.parallelism` 来设置 `ForkJoinPool` 线程池的线程数）。如果所有 `CompletableFuture` 共享一个线程池，那么一旦有任务执行一些很慢的 I/O 操作，就会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成线程饥饿，进而影响整个系统的性能。所以，强烈建议你要**根据不同的业务类型创建不同的线程池，以避免互相干扰**。 #### CompletableFuture 结果转换 | 方法 | 作用描述 | | :---------------- | :----------------- | | `thenApply` | 同步转换结果 | | `thenApplyAsync` | 异步转换结果 | | `thenAccept` | 同步消费结果 | | `thenAcceptAsync` | 异步消费结果 | | `thenRun` | 同步执行无参数操作 | | `thenRunAsync` | 异步执行无参数操作 | CompletableFuture 结果转换用法 ```java CompletableFuture future = CompletableFuture.supplyAsync(() -> "hello") .thenApply(s -> s + " world") // 同步转换 .thenApply(String::toUpperCase) // 继续转换 .thenApply(s -> s + "!"); System.out.println(future.get()); // 输出：HELLO WORLD! ``` #### CompletableFuture 组合 | 方法 | 作用描述 | | :--------------- | :----------------------- | | `thenCompose` | 链式组合（扁平化） | | `thenCombine` | 合并两个独立任务结果 | | `thenAcceptBoth` | 合并两个结果并消费 | | `runAfterBoth` | 两个任务都完成后执行操作 | CompletableFuture 组合用法 ```java // thenCompose - 链式组合 CompletableFuture future = getUserInfo(userId).thenCompose(user -> getOrderHistory(user)); // thenCombine - 合并两个独立任务结果 CompletableFuture future1 = CompletableFuture.supplyAsync(() -> "Hello"); CompletableFuture future2 = CompletableFuture.supplyAsync(() -> "World"); CompletableFuture combined = future1.thenCombine(future2, (s1, s2) -> s1 + " " + s2); ``` #### CompletableFuture 多任务组合 | 方法 | 作用描述 | | :------ | :----------------- | | `allOf` | 所有任务完成后执行 | | `anyOf` | 任意任务完成后执行 | CompletableFuture 多任务组合用法 ```java CompletableFuture task1 = CompletableFuture.supplyAsync(() -> "结果 1"); CompletableFuture task2 = CompletableFuture.supplyAsync(() -> "结果 2"); CompletableFuture task3 = CompletableFuture.supplyAsync(() -> "结果 3"); // 所有任务完成后执行 CompletableFuture all = CompletableFuture.allOf(task1, task2, task3); // 任意一个任务完成后执行 CompletableFuture